#!/bin/bash

set -e

echo "开始部署 Kafka KRaft 模式学习环境..."

# 准备 Docker 镜像
echo "0. 拉取 Docker 镜像... (如果已经存在可以跳过)"
pull_if_not_exists() {
    IMAGE_NAME=$1
    if docker image inspect $IMAGE_NAME >/dev/null 2>&1; then
        echo "Image $IMAGE_NAME already exists locally. Skipping pull."
    else
        echo "Image $IMAGE_NAME not found. Pulling..."
        docker pull $IMAGE_NAME
    fi
}

pull_if_not_exists bitnami/kafka:4.0.0
pull_if_not_exists provectuslabs/kafka-ui:v0.7.2
# pull_if_not_exists curlimages/curl:latest
pull_if_not_exists python:3.13-slim

# 创建 Kind 集群
echo "1. 创建 Kind 集群..."
kind create cluster --config deployment/00-kind-kafka-practise.yaml

# 预先加载镜像到 Kind 集群
echo "2. 加载基础镜像到 Kind 集群..."
echo "  加载 Kafka 镜像..."
kind load docker-image bitnami/kafka:4.0.0 --name kafka-practise

echo "  加载 Kafka-UI 镜像..."
kind load docker-image provectuslabs/kafka-ui:v0.7.2 --name kafka-practise

# 构建统一的应用镜像 - 使用明确的标签
echo "3. 构建统一的应用镜像..."
docker build -t localhost/kafka-app:0.0.1 -f app/Dockerfile app/

# 加载应用镜像到 Kind 集群 - 使用 Podman 兼容的方式
echo "4. 加载应用镜像到 Kind 集群..."
kind load docker-image localhost/kafka-app:0.0.1 --name kafka-practise

# 创建K8s命名空间
echo "5. 创建命名空间 kafka-practise..."
kubectl apply -f deployment/01-namespace.yaml

# 部署到 Kubernetes
echo "6. 部署到 Kubernetes..."

echo "  创建命名空间..."
kubectl apply -f deployment/01-namespace.yaml

echo "  部署 Kafka (KRaft 模式)..."
kubectl apply -f deployment/02-kafka.yaml

echo "  等待 Kafka 启动..."
sleep 30
kubectl wait --for=condition=ready pod -l app=kafka -n kafka-practise --timeout=300s

echo "  检查 Kafka 启动状态..."
kubectl logs -l app=kafka -n kafka-practise --tail=50

echo "  确保 Kafka 完全启动..."
sleep 30

echo "  部署 Kafka UI..."
kubectl apply -f deployment/03-kafka-ui.yaml

echo "  等待 Kafka UI 启动..."
sleep 20

echo "  部署 Producer..."
kubectl apply -f deployment/04-producer.yaml

echo "  部署 Consumer..."
kubectl apply -f deployment/05-consumer.yaml

# 等待所有 Pod 就绪
echo "7. 等待所有服务启动..."
sleep 30

# 显示部署状态
echo "8. 部署状态:"
kubectl get pods -n kafka-practise

echo "9. 测试 Kafka 连接..."
# 创建测试 Pod
kubectl create -f - <<EOF
apiVersion: v1
kind: Pod
metadata:
  name: kafka-test
  namespace: kafka-practise
spec:
  containers:
  - name: test
    image: localhost/kafka-app:0.0.1
    imagePullPolicy: IfNotPresent
    command: ["sleep", "infinity"]
  restartPolicy: Never
EOF

echo "等待测试 Pod 启动..."
sleep 10

echo "运行简单测试..."
kubectl exec kafka-test -n kafka-practise -- python -c "
from kafka import KafkaProducer
import json
try:
    producer = KafkaProducer(bootstrap_servers=['kafka-0.kafka.kafka-practise.svc.cluster.local:9092'])
    print('✅ Kafka 连接成功!')
except Exception as e:
    print(f'❌ Kafka 连接失败: {e}')
"

# 清理测试 Pod
kubectl delete pod kafka-test -n kafka-practise

echo "10. 服务访问信息:"
echo "  Kafka UI: http://localhost:30080"
echo "  Kafka Broker: localhost:30094"
echo ""
echo "部署完成!"
